package cn.doitedu.demo6;


import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.HashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 * <p>
 * 规则id: rule-004
 * 规则条件：
 * 动态画像统计条件
 * 添加购物车事件达到 4次以上
 * 且历次添加到购物车的商品的价格平均值>80.5
 * 触发条件：
 * 发生下单事件
    {
         "rule_id":"rule-m01-001",
         "target_event":"add_cart",
         "target_event_min_count":2,
         "target_property":"item_price",
         "property_avg_min":80.5,
         "trigger_event":"create_order"
    }
 * <p>
 * 规则id: rule-005
 * 规则条件：
 * 动态画像统计条件
 * 收藏商品事件达到 6次以上
 * 且历次收藏的商品的价格平均值>120
 * 触发条件：
 * 发生分享事件
 * <p>
    {
        "rule_id":"rule-m01-002",
        "target_event":"item_favor",
        "target_event_min_count":1,
        "target_property":"item_price",
        "property_avg_min":20,
        "trigger_event":"item_share"
    }
 **/

public class RuleModelCalculator_01 implements RuleCalculator {

    ValueState<Integer> cntState;
    ValueState<Double> sumState;

    JSONObject ruleParamJsonObject;

    JSONObject message;

    private String ruleId;

    @Override
    public void init(RuntimeContext runtimeContext, RuleMeta ruleMeta) throws IOException {

        String ruleParamJson = ruleMeta.getRule_param_json();

        ruleParamJsonObject = JSON.parseObject(ruleParamJson);
        ruleId = ruleParamJsonObject.getString("rule_id");

        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId() + "-cnt", Integer.class));
        sumState = runtimeContext.getState(new ValueStateDescriptor<Double>(ruleId() + "-sum", Double.class));

        message = new JSONObject();
        message.put("rule_id", ruleId);
    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        String eventId = userAction.getEvent_id();
        HashMap<String, String> properties = userAction.getProperties();

        // 从规则参数中取出目标事件
        String targetEvent = ruleParamJsonObject.getString("target_event");
        // 从目标事件中取目标属性
        String targetProperty = ruleParamJsonObject.getString("target_property");

        if (eventId.equals(targetEvent)) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);

            double propertyValue = Double.parseDouble(properties.get(targetProperty));
            sumState.update(sumState.value() == null ? propertyValue : sumState.value() + propertyValue);
        }


        // 从规则参数中，取触发行为事件id
        String triggerEvent = ruleParamJsonObject.getString("trigger_event");
        // 从规则参数中，取目标事件的发生次数参数
        int min_count = ruleParamJsonObject.getIntValue("target_event_min_count");
        // 从规则参数中，取目标事件的目标属性的平均值参数
        double propertyAvgMin = ruleParamJsonObject.getDoubleValue("property_avg_min");

        if (eventId.equals(triggerEvent)) {

            Integer cnt = cntState.value();
            Double sum = sumState.value();

            if (cnt != null && sum != null && cnt > min_count && sum / cnt > propertyAvgMin) {
                message.put("user_id", userAction.getUser_id());
                message.put("hit_time", userAction.getAction_time());

                out.collect(message.toJSONString());

            }

        }
    }

    @Override
    public String ruleId() {
        return this.ruleId;
    }


}
